package practise.mulitThreads.ConcurrentInPractice.BoundedBuffer;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by richard on 6/14/14.
 */
public class BoundedBuffer<T> {
    private Lock lockForPut = new ReentrantLock();
    private Lock lockForGet = new ReentrantLock();
    private static final int DEFAULT_CAPABILITY=10;
    private Condition emptyCondition = lockForGet.newCondition();
    private Condition fullCondition = lockForPut.newCondition();
    private Integer capability = new Integer(DEFAULT_CAPABILITY);
    private AtomicInteger size = new AtomicInteger(0);
    private List<T> container;
    public BoundedBuffer(int capability){
        this.capability = capability;
        container = new LinkedList<>();
    }
    public BoundedBuffer(){container = new LinkedList<>();}

    public void offer(T object){

        int c = 0;
        try{
            lockForPut.lock();
            c=size.get();
            if(c>=capability){
                waitWhenFull();
            }
            if(size.incrementAndGet()+1<capability){
                notifyForFull();
            }
            container.add(object);

        } catch (InterruptedException e) {
            //interrupted
        } finally {
            lockForPut.unlock();
        }
        if(c==0){
            notifyForEmpty();
        }

    }

    public T poll(){
        int c=size.get();
        T obj=null;
        try{
            lockForGet.lock();
            c=size.get();
            if(c==0){
                waitWhenEmpty();
            }
            if(size.decrementAndGet()-1>0){
                notifyForEmpty();
            }
            obj = container.remove(container.size()-1);
        } catch (InterruptedException e) {
            //
        } finally {
            lockForGet.unlock();
        }
        if(c==capability){
            notifyForFull();
        }
        return obj;
    }

    private void waitWhenFull() throws InterruptedException {
        System.out.println("wait for a place.");
        lockForPut.lock();
        try{
            fullCondition.await();
        }
        finally {
            lockForPut.unlock();
        }

    }

    private void notifyForFull(){

        lockForPut.lock();
        try {
            fullCondition.signal();
        }
        finally {
            lockForPut.unlock();
        }

    }

    private void waitWhenEmpty() throws InterruptedException {
        System.out.println("wait for a element.");
        lockForGet.lock();
        try{
            emptyCondition.await();
        }
        finally {
            lockForGet.unlock();
        }

    }

    private void notifyForEmpty(){

        lockForGet.lock();
        try {
            emptyCondition.signal();
        }
        finally {
            lockForGet.unlock();
        }

    }
}
